package club.baimahu.pencil.miaosha.rocketmq;


import java.io.UnsupportedEncodingException;
import java.util.Date;
import java.util.List;

import club.baimahu.pencil.miaosha.domain.Stock;
import club.baimahu.pencil.miaosha.domain.StockOrder;
import club.baimahu.pencil.miaosha.service.StockService;
import club.baimahu.pencil.miaosha.service.UserService;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

/**
 * This example shows how to subscribe and consume messages using providing {@link DefaultMQPushConsumer}.
 */


@Slf4j
@Component
public class Consumer implements InitializingBean {

    @Value("${rocketmq.consumer.namesrvAddr}")
    public String namesrvAddr;

    @Value("${rocketmq.consumer.groupName}")
    public String groupName;

    @Autowired
    StockService stockService;

    public DefaultMQPushConsumer startDecreaseStockConsumer() throws MQClientException {

        String topic = StockService.DECREASE_STOCK_TOPIC;
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);

        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.subscribe(topic, "*");
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.setInstanceName("consumer"+"-" +System.currentTimeMillis());
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                log.info("开始消费消息[{}],消息size={}",topic,msgs.size());
                for(MessageExt ext : msgs){
                    byte[] bytes = ext.getBody();
                    try {
                        String body = new String(bytes,"utf-8");
                        if(body != null){
                            Integer sid = Integer.parseInt(body);

                            int affect = stockService.decreaseStockFast(sid);
                            if(affect == 1){
                                log.info("MQ扣减库存成功");
                            }else{
                                log.error("MQ扣减库存失败,需要人工介入");
                                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                            }
                        }
                    } catch (UnsupportedEncodingException e) {
                        e.printStackTrace();
                    } catch (Exception e){
                        log.error("消费消息出现异常,message : {}" , e.getLocalizedMessage());
                        //可发送邮件短信进行相应报警
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();

        log.info("启动ROCKETMQ消费者，监听Topic:{}",topic);

        return  consumer;
    }

    public DefaultMQPushConsumer startCreateOrderConsumer() throws MQClientException {

        String topic = StockService.CREATE_STOCK_ORDER_TOPIC;

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);

        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.setInstanceName("consumer1"+"-" +System.currentTimeMillis());
        consumer.subscribe(topic, "*");
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.registerMessageListener(new MessageListenerConcurrently() {

            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                log.info("开始消费消息[{}],消息size={}",topic,msgs.size());
                for(MessageExt ext : msgs){
                    byte[] bytes = ext.getBody();
                    try {
                        String body = new String(bytes,"utf-8");
                        if(body != null){
                            String[] arr = body.split(";");
                            if(arr.length != 3 || !arr[0].matches("[0-9]+")){
                                log.warn("消息格式不符合规范,忽略该消息,body = {}",body);
                                continue;
                            }
                            Integer sid = Integer.parseInt(arr[0]);
                            Integer userId = Integer.parseInt(arr[2]);
                            String name = arr[1];

                            StockOrder stockOrder = new StockOrder();
                            stockOrder.setId(Long.parseLong(ext.getKeys()));
                            stockOrder.setSid(sid);
                            stockOrder.setName(name);
                            stockOrder.setUserId(userId);
                            stockOrder.setCreateTime(new Date());
                            stockOrder.setTxId(ext.getKeys());

                            int affect = stockService.createStockOrder(stockOrder);
                            if(affect == 1){
                                log.info("创建订单成功,stockOrder={}",stockOrder.toString());
                                //return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                            }else{
                                log.error("创建订单失败,stockOrder={}",stockOrder.toString());
                            }
                        }
                    } catch (UnsupportedEncodingException e) {
                        e.printStackTrace();
                    } catch (Exception e){
                        log.error("消费消息出现异常,message : {}" , e.getLocalizedMessage());
                        //可发送邮件短信进行相应报警
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();

        log.info("启动ROCKETMQ消费者，监听Topic:{}",topic);

        return  consumer;
    }


    @Override
    public void afterPropertiesSet() throws Exception {
        DefaultMQPushConsumer consumer1 = startCreateOrderConsumer();
        DefaultMQPushConsumer consumer2 = startDecreaseStockConsumer();
        registerHook(consumer1);
        registerHook(consumer2);
    }

    public void destroy(DefaultMQPushConsumer consumer){

        if(consumer != null){
            String instanceName = consumer.getInstanceName();
            log.info("start destroy RocketMQ  instance[{}]",instanceName);
            consumer.shutdown();
            log.info("finish destroy RocketMQ instance[{}]",instanceName);
        }
    }

    private void registerHook(DefaultMQPushConsumer consumer) {
        Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
            @Override
            public void run() {
                destroy(consumer);
            }
        }));
    }
}
